package spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test


class Transformation {

  //local模式（本地运行）
  val conf = new SparkConf().setMaster("local[*]").setAppName("word_count");
  val sc = new SparkContext(conf);

  ////////////////////单value类型//////////////////////
  @Test
  def MapTest(): Unit = {
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 10)
    val mapRDD: RDD[Int] = seqRDD.map(data => data * 10)
    mapRDD.collect().foreach(println)
  }

  @Test
  def MapPartitionsTest(): Unit = {
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 10)
    val mpsRDD: RDD[Int] = seqRDD.mapPartitions(x => x.map(_ * 2))
    mpsRDD.collect().foreach(println)
  }

  @Test
  def MapPartitionsWithIndexTest(): Unit = {
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 10)
    val mpwiRDD: RDD[(Int, Int)] = seqRDD.mapPartitionsWithIndex((index, items) => (items.map((index, _))))
    mpwiRDD.collect().foreach(println)
  }

  @Test
  def FlatMapTest(): Unit = {
    val listRDD: RDD[List[Int]] = sc.makeRDD(Array(List(1, 2), List(3, 4)))
    val flatMapRDD: RDD[Int] = listRDD.flatMap(datas => datas)
    flatMapRDD.collect().foreach(println)
  }

  @Test
  def GlomTest(): Unit = {
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 8, 3)
    val glomRDD: RDD[Array[Int]] = seqRDD.glom()
    sc.makeRDD(1 to 9)
    glomRDD.collect().foreach(array => {
      println(array.mkString(","))
    })
  }

  @Test
  def GroupByTest(): Unit = {
    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val groupByRDD: RDD[(Int, Iterable[Int])] = listRDD.groupBy(i => i % 2)
    groupByRDD.collect().foreach(println)
  }

  @Test
  def FilterTest(): Unit = {
    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val filterRDD: RDD[Int] = listRDD.filter(x => x % 2 == 0)
    filterRDD.collect().foreach(println)
  }

  //从指定的数据集合中进行抽样操作
  @Test
  def SampleTest(): Unit = {
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 10)
    val sampleRDD: RDD[Int] = seqRDD.sample(false, 0.4, 1)
    sampleRDD.collect().foreach(println)
  }

  //去重 （有shuffle阶段,所以不一定在原来的分区中）
  //将RDD中一个分区的数据打乱重组到其他不同的分区的操作，称之为shuffle操作
  @Test
  def DistinctTest(): Unit = {
    val listRDD: RDD[Int] = sc.makeRDD(List(1, 1, 2, 2, 3, 3, 4, 4, 5, 5))
    val distinctRDD: RDD[Int] = listRDD.distinct()
    //    使用distinct算子对数据去重，但是因为去重后会导致数据减少，所以可以改变默认的分区的数量
    //    val distinctRDD: RDD[Int] = listRDD.distinct(2)
    distinctRDD.collect().foreach(println)
  }

  //缩减分区
  //coalesce有可以设置是否shuffle的参数（默认是false），而repartition必须进行shuffle
  @Test
  def CoalesceTest(): Unit ={
    val seqRDD: RDD[Int] = sc.makeRDD(1 to 16, 4)
    println("缩减分区前的分区个数"+seqRDD.partitions.size)
    val coalesceRDD: RDD[Int] = seqRDD.coalesce(3)
    println("缩减分区后的分区个数"+coalesceRDD.partitions.size)
  }

  //排序
  @Test
  def SortByTest(): Unit ={
    val listRDD: RDD[Int] = sc.makeRDD(List(3, 1, 4, 2))
    val sortByAscRDD: RDD[Int] = listRDD.sortBy(x => x)
    sortByAscRDD.collect().foreach(println)
    val sortByDescRDD: RDD[Int] = listRDD.sortBy(x => x,false)
    println("----------")
    sortByDescRDD.collect().foreach(println)
  }

  ////////////////////双value类型//////////////////////
  //聚合
  @Test
  def ReduceByKeyTest(): Unit ={
    val wordsRDD: RDD[String] = sc.makeRDD(Array("one", "two", "three", "one"))
    val pairRDD: RDD[(String, Int)] = wordsRDD.map((_, 1))
    val ReduceByRDD: RDD[(String, Int)] = pairRDD.reduceByKey(_ + _)
    ReduceByRDD.collect().foreach(println)
  }

  @Test
  def GroupByKey(){}

  @Test
  def AggregateByTest(): Unit ={
    val listRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 1), ("c", 1), ("a", 1)),2)
    val aggregateByKeyRDD: RDD[(String, Int)] = listRDD.aggregateByKey(0)(math.max(_, _), _ + _)
    aggregateByKeyRDD.collect().foreach(println)
  }

  //
}
